package com.tlgen.coopera.runner;

import com.alibaba.fastjson.JSONObject;
import com.tlgen.coopera.model.BeatRegistryServer;
import com.tlgen.coopera.model.RegistryServer;
import com.tlgen.coopera.factory.RegistryLocalFactory;
import com.tlgen.coopera.service.RegistryService;
import com.tlgen.coopera.utils.DateUtils;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.DelayQueue;
import java.util.stream.Collectors;

/*
 *@title MyDelayTaskRunner
 *@description
 *@author ta151486
 *@version 1.0
 *@create 2023/12/14 9:46
 */
@Component
public class MyDelayTaskRunner implements InitializingBean {

    @Autowired
    private DelayQueue<MyDelayTask> delayQueue;

    @Autowired
    private RegistryService registryService;

    @Override
    public void afterPropertiesSet() {
        new Thread(() -> {
            try {
                while (true) {
                    MyDelayTask take = delayQueue.take();
                    BeatRegistryServer beatRegistryServer = take.getBeatRegistryServer();
                    System.out.println("执行延迟任务: " + JSONObject.toJSONString(beatRegistryServer));
                    List<BeatRegistryServer> beatRegistryList = RegistryLocalFactory.getInstance().getBeatList();
                    // 获取非健康节点
                    List<BeatRegistryServer> nonHealthyList = beatRegistryList.stream().filter(x -> {
                        if (Objects.isNull(x.getReceiveAckTime()))
                            return true;
                        long seconds = Duration.between(DateUtils.DateStringToLocalDateTime(x.getSendBeatTime()),
                                DateUtils.DateStringToLocalDateTime(x.getReceiveAckTime())).getSeconds();
                        return seconds > 6;
                    }).collect(Collectors.toList());
                    // 设置健康节点
                    if (CollectionUtils.isNotEmpty(nonHealthyList)) {
                        System.out.println("非健康节点: " + JSONObject.toJSONString(nonHealthyList));
                        beatRegistryList.removeAll(nonHealthyList);
                        System.out.println("剩下健康的节点: " + JSONObject.toJSONString(beatRegistryList));
                        List<String> ipList = nonHealthyList.stream().map(BeatRegistryServer::getIp).collect(Collectors.toList());
                        List<RegistryServer> registryList = RegistryLocalFactory.getInstance().getRegistryList();
                        RegistryLocalFactory.getInstance().setBeatList(beatRegistryList);
                        RegistryLocalFactory.getInstance().setRegistryList
                                (registryList.stream().filter(x -> !ipList.contains(x.getIp())).collect(Collectors.toList()));
                        // 将健康节点的服务列表复制到其他节点
                        registryService.notifyAllInstance();
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }).start();
    }

}
